<?php
declare(strict_types=1);

namespace app\common\service\rabbitMQ;

use app\common\exception\LogicException;
use app\common\model\MessageIdempotence;
use Exception;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Exchange\AMQPExchangeType;
use PhpAmqpLib\Wire\AMQPTable;
use think\facade\Env;

/**
 * rabbitMQ类
 * @package app\common\service
 */
class RabbitMQ
{
    #tcp连接（客户端->RabbitMQ）
    private $connection;
    #信道
    private $channel;
    #交换器名称
    private $exchange;
    /**
     * @var array
     * alternate-exchange https://cloud.tencent.com/developer/article/1529847
     */
    private $exchangeArguments = [];
    /**
     * 交换器类型
     * const DIRECT = 'direct'; 完全匹配
     * const FANOUT = 'fanout'; 发送到所有与之绑定的queue中
     * const TOPIC = 'topic'; 模式匹配
     * const HEADERS = 'headers';
     * @var string
     */
    private $exchangeType = AMQPExchangeType::DIRECT;
    #队列名称
    private $queue;
    #队列名称
    private $routingKey;
    /**
     * 被动模式
     * true 如果$queue不存在，则返回错误(不创建队列，只是检测队列是否存在)
     * false 如果$queue不存在，则创建此队列，然后返回OK
     * @var bool
     */
    private $passive = false;
    /**
     * 队列中的消息在被最后一个消费者消费后队列将自动删除，并且断开与其它消费者之间的连接。
     * @var bool
     */
    private $autoDelete = false;
    /**
     * 持久化,数据持久化到磁盘(e的mn中)，可以防止mq宕机重启数据丢失
     * exchange queue message三个都持久化才能使用
     * @var bool
     */
    private $durable = true;
    /**
     * 排他性，为true时，只能在本次连接中使用，连接关闭时自动消亡(即使$durable为true也是)
     * @var bool
     */
    private $exclusive = false;
    private $noAck     = false;
    /**
     * 异步执行，为true时，不等待队列创建结果，立即完成函数调用
     * @var bool
     */
    private $noWait = false;
    /**
     * @var array
     * x-message-ttl 设置5000，则表示消息超过5秒没被处理就会超时过期
     * x-expires 设置120000表示队列在2分钟内没被消费则被删除
     * x-max-length       表示传送数据最大长度
     * x-max-length-bytes 表示传送数据字节数
     * x-dead-letter-exchange     死信交换机
     * x-dead-letter-routing-key  死信路由
     * x-max-priority 表示设置优先级，范围为0~255，只有当消息堆积的时候，这个优先级才有意义，数字越大优先级越高
     * x-queue-mode 当为lazy，表示惰性队列，3.6.0之后才被引入的概念，相比默认的模式，惰性队列模式会将生产者产生的消息直接存到磁盘中，这当然会增加IO开销，但适合应对大量消息堆积的情况；因为当大量消息堆积时，内存也不够存放，会将消息转存到磁盘，这个过程也是比较耗时且过程中不能接收新的消息。如果需要将普通队列转换成惰性队列需要将原来的队列删除，重新创建个惰性队列绑定。
     *
     */
    private $queueArguments = [];
    private $ticket;
    /**
     * @var bool
     * 若为true,表示这个exchange不可以被client用来推送消息，仅用来进行exchange和exchange之间的绑定。
     */
    private $internal = false;
    #强制性的
    private $mandatory = false;
    #立即操作
    private $immediate = false;
    #消费者参数，每次派发多少数量的任务过来
    private $prefetchCount = 1;
    //-------------------------------- 延迟消费队列相关 start -------------------------
    #超时秒数
    private $xDelay;
    //-------------------------------- 延迟消费队列相关 end -------------------------
    /**
     * 发布确认模式(保证发布者一定发布成功)：
     * 1 单独发布消息，同步等待确认：简单，但吞吐量非常有限。
     * 2 批量发布消息，同步等待批量确认：简单，合理的吞吐量，但是很难推断出什么时候出了问题。
     * 3 异步处理：最佳性能和资源使用，在出现错误的情况下可以很好地控制，但可以正确实施。
     */
    private $confirmSelectType = false;
    private $confirmSelectSetAckHandler;
    private $confirmSelectSetNAckHandler;
    private $confirmSelectSetReturnListener;

    public function setExchange($v): RabbitMQ
    {
        $this->exchange = $v;
        return $this;
    }

    public function setExchangeArguments(array $v): RabbitMQ
    {
        $this->exchangeArguments = $v;
        return $this;
    }

    public function setExchangeType($v): RabbitMQ
    {
        $this->exchangeType = $v;
        return $this;
    }

    public function setQueue($v): RabbitMQ
    {
        $this->queue = $v;
        return $this;
    }

    public function setQueueArguments($v): RabbitMQ
    {
        $this->queueArguments = $v;
        return $this;
    }

    public function setRoutingKey($v): RabbitMQ
    {
        $this->routingKey = $v;
        return $this;
    }

    public function setPassive($v): RabbitMQ
    {
        $this->passive = $v;
        return $this;
    }

    public function setAutoDelete($v): RabbitMQ
    {
        $this->autoDelete = $v;
        return $this;
    }

    public function setDurable($v): RabbitMQ
    {
        $this->durable = $v;
        return $this;
    }

    public function setExclusive($v): RabbitMQ
    {
        $this->exclusive = $v;
        return $this;
    }

    public function setNoAck($v): RabbitMQ
    {
        $this->noAck = $v;
        return $this;
    }

    public function setNoWait($v): RabbitMQ
    {
        $this->noWait = $v;
        return $this;
    }

    public function setTicket($v): RabbitMQ
    {
        $this->ticket = $v;
        return $this;
    }

    public function setInternal($v): RabbitMQ
    {
        $this->internal = $v;
        return $this;
    }

    public function setMandatory($v): RabbitMQ
    {
        $this->mandatory = $v;
        return $this;
    }

    public function setImmediate($v): RabbitMQ
    {
        $this->immediate = $v;
        return $this;
    }

    public function setPrefetchCount($v): RabbitMQ
    {
        $this->prefetchCount = $v;
        return $this;
    }

    //    public function setApplicationHeaders(array $v): RabbitMQ
    //    {
    //        $this->applicationHeaders = $v;
    //        return $this;
    //    }

    public function setXDelay(int $v): RabbitMQ
    {
        $this->xDelay = $v;
        return $this;
    }

    /**
     * 消息确认模式
     * @param $v
     * @param array $funcArr
     * @return RabbitMQ
     */
    public function setConfirmSelectType($v, $funcArr = []): RabbitMQ
    {
        $this->confirmSelectType = $v;
        if (isset($funcArr[0])) {
            $this->confirmSelectSetAckHandler = $funcArr[0];
        }
        if (isset($funcArr[1])) {
            $this->confirmSelectSetNAckHandler = $funcArr[1];
        }
        if (isset($funcArr[2])) {
            $this->confirmSelectSetReturnListener = $funcArr[2];
        }
        return $this;
    }

    public function __construct()
    {
        try {
            $this->connection = new AMQPStreamConnection(
                Env::get('rabbitmq.host'),
                Env::get('rabbitmq.port'),
                Env::get('rabbitmq.user'),
                Env::get('rabbitmq.pass'),
                Env::get('rabbitmq.vhost')
            );
            $this->channel    = $this->connection->channel();
        } catch (Exception $e) {
            throw new LogicException(['msg' => $e->getMessage()]);
        }
    }

    /**
     * @Todo 一定要将所有的对象成员全部赋值后再执行这个方法，最后生产
     * @return RabbitMQ
     */
    public function preparePublisher(): RabbitMQ
    {
        $this->prepare();
        if ($this->confirmSelectType) {
            $this->channel->confirm_select();
            switch ($this->confirmSelectType) {
                case 1:
                case 2:
                    break;
                case 3:
                    $this->channel->set_ack_handler($this->confirmSelectSetAckHandler);
                    $this->channel->set_nack_handler($this->confirmSelectSetNAckHandler);
                    $this->channel->set_return_listener($this->confirmSelectSetReturnListener);
                    break;
                default:
            }
        }
        return $this;
    }

    public function publisher($argvArr): void
    {
        switch ($this->confirmSelectType) {
            case 1:
                foreach ($argvArr as $item) {
                    $this->publisherMain($item);
                    //会报错Server ack'ed unknown delivery_tag "2"
                    //阻塞等待消息确认
                    $this->channel->wait_for_pending_acks(5);
                }
                break;
            case 2:
                foreach ($argvArr as $item) {
                    $this->publisherMain($item);
                    //会报错Server ack'ed unknown delivery_tag "2"
                    //阻塞等待消息确认
                }
                $this->channel->wait_for_pending_acks(5);
                break;
            case 3:
            default:
                foreach ($argvArr as $item) {
                    $this->publisherMain($item);
                    //会报错Server ack'ed unknown delivery_tag "2"
                    //阻塞等待消息确认
                }
                $this->channel->wait_for_pending_acks_returns();
                break;
        }
    }

    private function publisherMain($item): void
    {
        /**
         * #消息持久化   AMQPMessage::DELIVERY_MODE_PERSISTENT
         * #消息非持久化 AMQPMessage::DELIVERY_MODE_NON_PERSISTENT
         */
        $messageId = getRandomStr(34);
        $message   = new AMQPMessage(
            serialize($item), [
                                'content_type'  => 'text/plain',
                                'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT,
                                'message_id'    => $messageId,
                                #投递模式
                            ]
        );

        if ($this->xDelay) {
            $message->set('application_headers', new AMQPTable(['x-delay' => $this->xDelay]));
        }

        $this->channel->basic_publish(
            $message,
            $this->exchange,
            $this->routingKey,
            $this->mandatory,
            $this->immediate,
            $this->ticket
        );
    }

    /**
     * 消费者
     * @param $consumerTag
     * @param $callback
     * @param bool $no_local
     * @throws \ErrorException
     */
    public function consumer($consumerTag, $callback, $no_local = false): void
    {
        /**
         * 限制每次只派发100个任务再进行下一次操作
         * $prefetch_size = null 此参数未使用
         * $prefetch_count = 100
         * $a_global = null 此参数未使用
         */
        $this->channel->basic_qos(null, $this->prefetchCount, null);

        /**
         * consumerTag 消费者标识符
         * no_local 不接收此使用者发布的消息
         * noAck 不需要手动确认
         * nowait 不等待
         * callback 回调方法
         */
        $this->channel->basic_consume(
            $this->queue,
            $consumerTag,
            $no_local,
            $this->noAck,
            $this->exclusive,
            $this->noWait,
            $callback,
            $this->ticket,
            new AMQPTable($this->queueArguments)
        );
        //
        //        /**
        //         * PHP终止时执行
        //         */
        //        register_shutdown_function(
        //            static function ($channel, $connection) {
        //                $channel->close();
        //                $connection->close();
        //            },
        //            $this->channel,
        //            $this->connection
        //        );

        //只要通道已注册回调,就循环
        while ($this->channel->is_consuming()) {
            $this->channel->wait();
        }
    }

    /**
     * @throws Exception
     */
    public function __destruct()
    {
        $this->channel->close();
        $this->connection->close();
    }

    public function prepare()
    {
        $this->channel->exchange_declare(
            $this->exchange,
            $this->exchangeType,
            $this->passive,
            $this->durable,
            $this->autoDelete,
            $this->internal,
            $this->noWait,
            new AMQPTable($this->exchangeArguments),
            $this->ticket
        );

        $this->channel->queue_declare(
            $this->queue,
            $this->passive,
            $this->durable,
            $this->exclusive,
            $this->autoDelete,
            $this->noWait,
            new AMQPTable($this->queueArguments),
            $this->ticket
        );

        $this->channel->queue_bind(
            $this->queue,
            $this->exchange,
            $this->routingKey,
            $this->noWait,
            new AMQPTable($this->queueArguments),
            $this->ticket
        );
    }

    /**
     * 幂等性操作，解决重复消费的问题
     * https://help.aliyun.com/document_detail/177412.html
     */
    public static function messageIdempotence($message)
    {
        $body       = unserialize($message->body);
        $properties = $message->get_properties();
        $routingKey = $message->getRoutingKey();
        $exchange   = $message->getExchange();
        try {
            $model              = new MessageIdempotence();
            $model->message_id  = $properties['message_id'];
            $model->routing_key = $routingKey;
            $model->properties  = $properties;
            $model->exchange    = $exchange;
            $model->body        = $body;
            $model->save();
            var_dump('未被消费，允许消费');
            return $model;
        } catch (\Exception $e) {
            //已消费，则直接返回OK
            var_dump('已被消费');
            return false;
        }
    }

    public static function ok($message): void
    {
        $message->ack();
        //        $message->delivery_info['channel']->basic_ack($message->delivery_info['delivery_tag']);
        //        if ($message->body === 'quit') {
        //            $message->getChannel()->basic_cancel($message->getConsumerTag());
        //        }
    }
}